package bigdata.scala.spark.streaming.ycfx.userstatusqueue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util

import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.HBaseClient
import bigdata.scala.spark.streaming.ycfx.model.UserStatus
import bigdata.scala.spark.streaming.ycfx.util.VoltageToHbUtil
import com.alibaba.fastjson.JSON
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{Put, Result}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.api.java.JavaInputDStream
import org.apache.spark.streaming.dstream.DStream
/**
 * @author : lubohua
 * @date : 2020/1/13 16:19
 * @description : 异常电压处理
 * 参数示例 : appname: UserStatus topics:Data1.1 groupId: duration: maxnum:
 */
object UserStatusQueue {
  def main(args: Array[String]): Unit = {
    val simpleSparkKafka = new SimpleSparkKafka(args)
    simpleSparkKafka.execute(new SKProcessor {
      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        val baseStream: DStream[String] = stream.map(cr=>cr.value())
        dataShareQueue(baseStream)
      }
    }).start()
  }

  //数据转换 将从kafka获取到的json字符串转为string类型
  def dataShareQueue(baseStream: DStream[String]): Unit ={
    baseStream.foreachRDD(rdd=>{
      val baseData: RDD[UserStatus] = rdd.map(kafka => JSON.parseObject(kafka))
        .map(json => {
          val meterId = json.getString("METER_ID")
          val meterAssetno = json.getString("METER_ASSETNO")
          val dataDate = json.getString("DATA_DATE")
          val phase = json.getString("PHASE")
          val dotdensity = json.getString("DOTDENSITY")
          val u1 = json.getString("U1")
          val u5 = json.getString("U5")
          val u9 = json.getString("U9")
          val u13 = json.getString("U13")
          val u17 = json.getString("U17")
          val u21 = json.getString("U21")
          val u25 = json.getString("U25")
          val u29 = json.getString("U29")
          val u33 = json.getString("U33")
          val u37 = json.getString("U37")
          val u41 = json.getString("U41")
          val u45 = json.getString("U45")
          val u49 = json.getString("U49")
          val u53 = json.getString("U53")
          val u57 = json.getString("U57")
          val u61 = json.getString("U61")
          val u65 = json.getString("U65")
          val u69 = json.getString("U69")
          val u73 = json.getString("U73")
          val u77 = json.getString("U77")
          val u81 = json.getString("U81")
          val u85 = json.getString("U85")
          val u89 = json.getString("U89")
          val u93 = json.getString("U93")
          UserStatus(meterId, meterAssetno, dataDate, dotdensity, phase, u1, u5, u9, u13, u17, u21, u25, u29, u33, u37, u41, u45, u49, u53, u57, u61, u65, u69, u73, u77, u81, u85, u89, u93)
        })
      val userMap: Map[String, String] = findTgUser(baseData)
      touchDown(userMap,baseData)
    })
  }

  //落地hbase
  def touchDown(userMap:Map[String,String],baseData:RDD[UserStatus]): Unit ={
    baseData.foreachPartition(data=>{

      data.foreach(t=>{
        val time1: LocalDateTime = LocalDateTime.parse(t.dataDate,DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
        val str = time1.toString.substring(0,4)
        val str1 = time1.toString.substring(5,7)
        val str2 = time1.toString.substring(8,10)
        val str3 = time1.toString.substring(11,13)
        val time = str+str1+str2+str3

        val rowkey:String = t.meterId+userMap.map(_._1)+"_"+userMap.map(_._2)+"_"+time
        val tableName:String = TableName.valueOf("").toString
        val columnFamily:String = ""

//        t.phase match {
//          case "u1" =>
//          case
//        }

        val dataMaps: util.HashMap[String, String] = VoltageToHbUtil.dataToMap(rowkey,t)

        HBaseClient.getInstance().put(tableName,rowkey,columnFamily,dataMaps)
      })
    })
  }

  //通过meterId查询hbase基表
  def findTgUser(baseData:RDD[UserStatus]): Map[String,String] ={
    var tgName:String = null
    var userName:String = null
    baseData.foreachPartition(data=>{
      data.foreach(t=>{
        val result: Result = HBaseClient.getInstance().get("dwm_user_meter_info",s"${t.meterId}")
        val strings: util.Set[String] = HBaseClient.getInstance().getRowkeyByRowFilter("","")

        tgName = ""
        userName = ""
      })
    })
    Map[String,String](tgName,userName)
  }

}
